﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Reflection;
using System.Runtime.Serialization.Formatters.Binary;
using System.Threading;
using System.Threading.Tasks;
using System.Xml.Linq;
using LucyGrid.Interface;
using LucyGrid.Messaging;
using LucyGrid.Nodes;

namespace LucyGrid
{
	/// <summary>
	/// Executes distributed process
	/// </summary>
	public class GridExecutor
	{
		/// <summary>
		/// Distributed image
		/// </summary>
		Object executorInstance;
		/// <summary>
		/// Image entry point
		/// </summary>
		MethodInfo executorMethod;
		/// <summary>
		/// Input object type
		/// </summary>
		Type executorMethodEntryParameterType;
		/// <summary>
		/// Grid endpoint
		/// </summary>
		string endpoint;

		static GridComm comm;

		private ILucyMaster imaster;


		/// <summary>
		/// Dummy data to force open connection 
		/// </summary>
		public readonly static Byte[] _startFrame;

		static GridExecutor()
		{
			_startFrame = new byte[] { 0xFF };
			comm = new GridComm();
		}

		/// <summary>
		/// Builds an executor for an endpoint
		/// </summary>
		/// <param name="endpoint"></param>
		public GridExecutor(string endpoint, ILucyMaster imaster)
		{
			try
			{
				Thread.CurrentThread.Name = string.Format("EXECUTOR REMOTE-{0}", Thread.CurrentThread.ManagedThreadId);
				GridUtils.NotifyLog("EXECUTOR,INFO", "Running REMOTE");
				this.imaster = imaster;
				this.endpoint = endpoint;
				LoadAssemblyInfo();
				ProcessChunks();
				GridUtils.NotifyLog("EXECUTOR,INFO", "Stopping REMOTE");
			}
			catch (Exception ex)
			{
				EmitError(ex.ToString());
				GridUtils.NotifyLog("EXECUTOR,INFO", "Error on REMOTE: {0}", ex.ToString());
				throw ex;
			}

		}

		public GridExecutor(Object executorInstance, MethodInfo executorMethod, string endpoint, ILucyMaster imaster)
		{
			try
			{
				Thread.CurrentThread.Name = string.Format("EXECUTOR DIRECT-{0}", Thread.CurrentThread.ManagedThreadId);
				LucyBinManager.imaster = imaster;
				GridUtils.NotifyLog("EXECUTOR,INFO", "Running DIRECT");
				this.endpoint = endpoint;
				// LoadAssemblyInfo();
				this.executorInstance = executorInstance;
				this.executorMethod = executorMethod;
				this.executorMethodEntryParameterType = executorMethod.GetParameters().First().ParameterType;
				ProcessChunks();
				GridUtils.NotifyLog("EXECUTOR,INFO", "Stopping DIRECT");
			}
			catch (Exception ex)
			{
				EmitError(ex.ToString());
				GridUtils.NotifyLog("EXECUTOR,INFO", "Error on DIRECT: {0}", ex.ToString());
				throw ex;
			}

		}

		/// <summary>
		/// Sends an error on the communication channel
		/// </summary>
		/// <param name="message"></param>
		private void EmitError(string message)
		{
			comm.EmitLog(message, true);
		}

		/// <summary>
		/// Gets a temp directory where to place distributed binary
		/// </summary>
		/// <returns></returns>
		public string GetTemporaryDirectory()
		{
			string tempDirectory = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName());
			Directory.CreateDirectory(tempDirectory);
			return tempDirectory;
		}

		/// <summary>
		/// Loads distributed image
		/// </summary>
		private bool LoadAssemblyInfo()
		{
			// Create temp directory
			var dir = GetTemporaryDirectory();
			GridUtils.NotifyLog("EXECUTOR,INFO", "Loading grid settings");
			// Load settings
			var settings = new GridComm.GridWebClient().DownloadString(endpoint + "/settings");
			var xml = XElement.Parse(settings);
			GridUtils.NotifyLog("EXECUTOR,INFO", "Loadding assemblies");
			// Files
			var assemblies = LucyBinManager.GetAssembliesFromFiles(xml.Descendants("file")
				.AsParallel()
				.Select(f =>
				{
					var tmpfile = dir + "\\" + f.Value;
					try
					{
						if (!Directory.Exists(Path.GetDirectoryName(tmpfile)))
							Directory.CreateDirectory(Path.GetDirectoryName(tmpfile));
						new GridComm.GridWebClient().DownloadFile(endpoint + "/file/" + f.Value, tmpfile);
						GridUtils.NotifyLog("EXECUTOR,INFO", "New file downloaded: {0}", tmpfile);
						return tmpfile.ToLower();
					}
					catch (Exception ex)
					{
						GridUtils.NotifyLog("EXECUTOR,INFO", "Binary file not found {0} on the master server. Error {1}", tmpfile, ex.ToString());
					}
					return null;
				})
				// Load references
				.Where(file => file != null && (file.EndsWith(".exe") || file.EndsWith(".dll")))
				.ToArray());
			// Executor type
			var exetype = xml.Descendants("exetype").First().Value;
			GridUtils.NotifyLog("EXECUTOR,INFO", "Loading executor Type {0}", exetype);
			Type executorType = assemblies.Select(a => a.GetTypes().FirstOrDefault(t => t.FullName == exetype)).First(t => t != null);
			// Executor method
			var method = xml.Descendants("exemethod").First().Value;
			GridUtils.NotifyLog("EXECUTOR,INFO", "Loading executor method {0}", method);
			executorMethod = executorType.GetMethods(BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance | BindingFlags.Static)
				.First(m => m.ToString() == method);
			executorMethodEntryParameterType = executorMethod.GetParameters().First().ParameterType;
			// Create the executor type
			executorInstance = LucyBinManager.CreateInstance(executorType);
			GridUtils.NotifyLog("EXECUTOR,INFO", "LoadAsemblyInfo done");
			return true;
		}

		/// <summary>
		/// Process chunks until an empty response is returned by the grid
		/// </summary>
		private void ProcessChunks()
		{
			var threads = Enumerable.Range(0, GridPerformance.NumberOfThreadsPerMachine);
			var tasks = threads
				.Select(i => GridTask.Run(() =>
				{
					Thread.CurrentThread.Name = string.Format("EXECUTOR CHUNKPROCESSOR-{0}", Thread.CurrentThread.ManagedThreadId);
					GridUtils.NotifyLog("EXECUTOR,DEBUG", "GridExecutor. Chunk processor thread start <{0}>",
						Thread.CurrentThread.ManagedThreadId.ToString());
					try
					{
						ChunkProcessor();
						GridUtils.NotifyLog("EXECUTOR,DEBUG", "GridExecutor. Chunk processor thread end");
					}
					catch (Exception ex)
					{
						GridUtils.NotifyLog("EXECUTOR,ERROR", "GridExecutor. Chunk processor thread ended with error: {0}", ex.ToString());
					}
				}))
				.ToArray();
			var heartbeat = new Heartbeat()
			{
				NodeId = GridComm.Id,
				WorkerId = Thread.CurrentThread.ManagedThreadId.ToString(),
				GridEndpoint = this.endpoint
			};
			while (!GridTask.WaitAll(tasks, 1000))
			{
				GridUtils.NotifyLog("EXECUTOR,INFO", "Sending heardbeat"); 
				comm.EmitHeartbeat(heartbeat);
				comm.EmitLog(string.Format("Processing chunks with {0} threads", System.Diagnostics.Process.GetCurrentProcess().Threads.Count), false);
			}
			comm.EmitLog("Processing chunks finalized", false);
			GridUtils.NotifyLog("EXECUTOR,INFO", "Processing chunks finalized"); 
				
		}

		void ChunkProcessor()
		{
			Chunk chunk = null;
			try
			{
				while ((chunk = GetNexChunk(chunk)) != null)
				{
					ProcessChunk(chunk);
				}
			}
			catch (Exception ex)
			{
				GridUtils.NotifyLog("EXECUTOR,ERROR", "Processing chunks error {0}", ex.ToString());
			}
		}

		/// <summary>
		/// Push/Pull chunks
		/// </summary>
		/// <param name="chunk"></param>
		/// <returns></returns>
		Chunk GetNexChunk(Chunk chunk)
		{
			if(chunk == null)
				CheckOverload();
			var overloaded = GridPerformance.IsHighLimitReached();
			var url = endpoint  + "/chunk";
			var send = chunk != null;
			var method = string.Empty;
			// Write if there's a chunk to return 
			if (send)
				method += "POST";
			// Read on the same operation
			if (!overloaded)
				method += "GET";
			if (send)
			{
				url += "/" + chunk.Head.Id;
			}
			var rq = (HttpWebRequest)WebRequest.Create(url);
			rq.Method = method;
			rq.AutomaticDecompression = DecompressionMethods.GZip | DecompressionMethods.Deflate;
			rq.Headers.Add("id", GridComm.Id);
			if (send)
			{
				rq.Headers.Add("isError", chunk.Head.IsError.ToString());
				using (var istream = rq.GetRequestStream())
				{
					// Reset input data before!
					chunk.Serialize(istream);
				}
			}
			if (!overloaded)
			{
				using (var rs = (HttpWebResponse) rq.GetResponse())
				{
					using (var ostream = rs.GetResponseStream())
					{
						if (rs.StatusCode == HttpStatusCode.NoContent)
						{
							chunk = null;
						}
						else
						{
							chunk = new Chunk(ostream, this.executorMethodEntryParameterType);
							chunk.Trace("Chunk received by executor");
						}
					}
				}
			}
			else
			{
				using (rq.GetResponse())
				{
				}
				chunk = GetNexChunk(null);
			}
			return chunk;
		}

		Chunk ProcessChunk(Chunk chunk)
		{
			GridUtils.NotifyLog("EXECUTOR,CHUNK,DEBUG", ", {0}, STEP, 2.1, Chunk processing start", chunk.Head.Id);
			// Process
			try
			{
				chunk.Trace("Processing start");
				chunk.Data = executorMethod.Invoke(executorInstance, new object[] { chunk.Data });
			}
			catch (Exception ex)
			{
				chunk.Data = ex.Message;
				chunk.Trace("Process error");
				GridUtils.NotifyLog("EXECUTOR,ERROR", "Processing chunk {0} error {1}", chunk.Head.Id, ex.ToString());
			}
			GridUtils.NotifyLog("EXECUTOR,CHUNK,DEBUG", ", {0}, STEP, 2.2, Chunk processing end", chunk.Head.Id);
			return chunk;
		}

		/// <summary>
		/// Chunk processor
		/// </summary>
		/// <param name="inputType"></param>
		/// <param name="serializer"></param>
		/// <returns></returns>
		/*[Obsolete]
		private bool ProcessChunk()
		{
			var sw = System.Diagnostics.Stopwatch.StartNew();
			long deSerial = 0, exec = 0, serial = 0;
			// Check before running
			CheckOverload();

			var parameters = new object[1];
			string chunkid = null;
			Exception httpex = null;
			try
			{
				// READ
				GridUtils.NotifyLog("EXECUTOR,DEBUG", "GridExecutor. Waiting for a chunk");
				if (!GridComm.Http(endpoint + "/chunk",
					null,
					null,
					(http, rs) =>
					{
						chunkid = http.Headers["chunkid"];
						// Sleep a bit until next chunk
						if (!string.IsNullOrEmpty(chunkid))
						{
							var dsw = System.Diagnostics.Stopwatch.StartNew();
							GridUtils.NotifyLog("EXECUTOR,DEBUG", "GridExecutor. Chunk received <{0}>", chunkid);
							parameters[0] = GridUtils.Deserialize(rs, executorMethodEntryParameterType);
							deSerial = dsw.ElapsedMilliseconds;
						}
					},
					(ex) =>
					{
						httpex = ex;
						GridUtils.NotifyLog("EXECUTOR,ERROR", "Error getting new chunk <{0}>", ex.ToString());
					}))
					throw httpex;
				// Sleep a bit until next chunk
				if (string.IsNullOrEmpty(chunkid))
				{
					// comm.EmitLog("No chunk received, sleeping a while", false);
					Thread.Sleep(100);
					return true;
				}
				GridUtils.NotifyLog("EXECUTOR,CHUNK,DEBUG", ", {0}, STEP, 2.1, Chunk processing start", chunkid);
				Object output = null;
				var isError = false;
				var ms = new MemoryStream();
				// Process
				try
				{
					var dsw = System.Diagnostics.Stopwatch.StartNew();
					output = executorMethod.Invoke(executorInstance, parameters);
					exec = dsw.ElapsedMilliseconds - deSerial;
					GridUtils.Serialize(ms, output);
					ms.Seek(0, SeekOrigin.Begin);
					serial = dsw.ElapsedMilliseconds - exec;
				}
				catch (Exception ex)
				{
					output = ex.Message;
					isError = true;
				}
				GridUtils.NotifyLog("EXECUTOR,CHUNK,DEBUG", ", {0}, STEP, 2.2, Chunk processing end, {1}", chunkid, output);
				// Write
				GridComm.Http(endpoint + "/chunk/" + chunkid,
					(http) =>
					{
						http.Method = "POST";
						http.Headers.Add("chunkid", chunkid);
						http.Headers.Add("isError", isError.ToString());
						http.Headers.Add("processTime", string.Format("DeSerial: {0}, Exec: {1}, Serial:{2}. Total: {3}",
								deSerial,
								exec,
								serial,
								sw.ElapsedMilliseconds.ToString()));
					},
					(http, s) =>
					{
						// comm.EmitLog("Start writting chunk response for " + chunkid, false);
						// Write
						//GridUtils.Serialize(s, output);
						ms.CopyTo(s);
						// comm.EmitLog("End writting chunk response for " + chunkid, false);
					},
					(http, s) =>
					{
						// comm.EmitLog("Processing chunk finalized " + chunkid + " " + s.ReadByte(), false);
					},
					(ex) =>
					{
						httpex = ex;
						// comm.EmitLog("Processing chunk finalized " + chunkid + " with error " + httpex.ToString(), true);
					});
				if (httpex != null)
					throw httpex;
				GridUtils.NotifyLog("EXECUTOR,CHUNK,DEBUG", ", {0}, STEP, 2.3, Chunk processing sent", chunkid);
			}
			catch (Exception ex)
			{
				// Release chunk
				if (!string.IsNullOrEmpty(chunkid))
				{
					GridUtils.NotifyLog("EXECUTOR,CHUNK,ERROR", ", {0}, STEP, 2.4, Chunk processing end with error, {1}", chunkid, ex.Message);
					GridComm.Http(endpoint + "/chunk/error/" + chunkid,
						(webreq) => webreq.Headers.Add("error", ex.Message));
				}
				return false;
			}
			return true;
		}*/

		/// <summary>
		/// Checks machine performance before processing.
		/// This method avoid overload the grid node.
		/// Sleep until performance conditions allows the node to work.
		/// <remarks>GC.Collect added to help releasing resources.
		/// http://stackoverflow.com/questions/478167/when-is-it-acceptable-to-call-gc-collect</remarks>
		/// </summary>
		private void CheckOverload()
		{
			if (GridPerformance.IsHighLimitReached())
			{
				do
				{
					GridUtils.NotifyLog("PERFORMANCE,INFO", "Node overloaded, waiting...");
					comm.EmitLog("Machine overload: Sleeping", true);
					Thread.Sleep(1000);
					try
					{
						GC.Collect();
						GC.WaitForPendingFinalizers();
						GC.Collect();
					}
					catch (Exception ex)
					{
						comm.EmitLog("Machine overload, GC.Collect error " + ex.ToString(), true);
					}
				} while (GridPerformance.IsLowLimitReached());
			}
		}
	}
}
